package cn.itcast.streaming

import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object UpdateStateByKey {

  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[6]").setAppName("update state by key")
    val sc = new SparkContext(sparkConf)
    sc.setLogLevel("ERROR")
    val ssc = new StreamingContext(sc, Seconds(1))

    val lines = ssc.socketTextStream(
      hostname = "192.168.88.100",
      port = 8888,
      storageLevel = StorageLevel.MEMORY_AND_DISK_SER
    )

    val words = lines.flatMap(_.split(" ")).map((_, 1))

    ssc.checkpoint("checkpoint")  // 使用updateStateByKey必须设置Checkpoint目录

    def updateFunc(newValue: Seq[Int], runningValue: Option[Int]) = {
      val currentBatchSum = newValue.sum
      val state = runningValue.getOrElse(0)
      Some(currentBatchSum + state)  // 返回的这个Some，会再次进入Checkpoint中当做状态存储
    }

    val wordCounts = words.updateStateByKey(updateFunc)

    wordCounts.print()

    ssc.start()
    ssc.awaitTermination()

  }

}
